package com.ctshk.rpc.bus.service.impl;


import java.io.IOException;
import java.time.LocalDate;
import java.time.LocalDateTime;
import java.time.LocalTime;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Predicate;
import java.util.stream.Collectors;

import com.ctshk.common.utils.EntityUtil;
import com.ctshk.rpc.bus.entity.es.BusLineCache;
import org.apache.dubbo.config.annotation.DubboReference;
import org.apache.dubbo.config.annotation.DubboService;
import org.apache.lucene.search.join.ScoreMode;
import org.elasticsearch.action.search.SearchRequest;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.index.query.BoolQueryBuilder;
import org.elasticsearch.index.query.QueryBuilders;
import org.elasticsearch.index.query.RangeQueryBuilder;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchHits;
import org.elasticsearch.search.builder.SearchSourceBuilder;
import org.elasticsearch.search.sort.FieldSortBuilder;
import org.elasticsearch.search.sort.SortOrder;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;

import com.alibaba.fastjson.JSON;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.ctshk.common.constant.ElasticSearchConstant;
import com.ctshk.common.constant.RedisConstants;
import com.ctshk.common.dto.Result;
import com.ctshk.common.elasticsearch.config.ElasticSearchConfiguration;
import com.ctshk.common.enums.SystemError;
import com.ctshk.common.enums.bus.BusRecommendationStatusCode;
import com.ctshk.common.exception.BusinessException;
import com.ctshk.common.req.OperationByLikeReq;
import com.ctshk.common.service.support.BaseService;
import com.ctshk.common.utils.Assist;
import com.ctshk.rpc.bus.dto.es.ArriveCityDTO;
import com.ctshk.rpc.bus.dto.es.BusLineCacheDTO;
import com.ctshk.rpc.bus.dto.es.BusProductCacheDTO;
import com.ctshk.rpc.bus.dto.es.BusStopsCacheCityDTO;
import com.ctshk.rpc.bus.dto.es.BusStopsCacheDTO;
import com.ctshk.rpc.bus.entity.BusProduct;
import com.ctshk.rpc.bus.mapper.BusElasticSearchMapper;
import com.ctshk.rpc.bus.mapper.BusProductMapper;
import com.ctshk.rpc.bus.mapper.BusStopsMapper;
import com.ctshk.rpc.bus.req.BusQueryReq;
import com.ctshk.rpc.bus.req.BusStopsDictListQueryReq;
import com.ctshk.rpc.bus.service.IBusLineService;
import com.ctshk.rpc.bus.service.IBusProductAppService;
import com.ctshk.rpc.bus.service.IBusProductService;
import com.ctshk.rpc.order.bus.service.IBusOrderService;

import lombok.extern.slf4j.Slf4j;

/**
 * <p>
 * 巴士订单表 服务实现类
 * </p>
 *
 * @author 彭柳霖
 * @since 2021-02-24
 */
@Slf4j
@DubboService
public class BusProductAppServiceImpl extends ServiceImpl<BusProductMapper, BusProduct> implements IBusProductAppService, BaseService {

    @DubboReference
    IBusOrderService busOrderService;

    @DubboReference
    IBusProductService busProductService;

    @DubboReference
    IBusLineService busLineService;

    @Autowired
    RestHighLevelClient restHighLevelClient;

    @Autowired
    RedisTemplate<String, String> redisTemplate;

    @Autowired
    BusElasticSearchMapper busElasticSearchMapper;
    
    @Autowired
    BusStopsMapper busStopsMapper;

    /**
     * 自由行 - 巴士 - 缓存 -初始化
     *
     * @return
     */
    @Override
    public Result busCacheInit() {
        // 清除旧数据
        busElasticSearchMapper.deleteAll();

        // 构建上架数据
        List<BusLineCacheDTO> busLineCacheDTOS = buildData(null, null);

        // 上架
        up(busLineCacheDTOS, RedisConstants.BUS_UPDATE_TIME);
        
        //同步站点主数据
        synStopsData();
        
        return Result.success();
    }

    /**
     * 构建数据
     *
     * @return
     */
    private List<BusLineCacheDTO> buildData(List<Long> ids, LocalDateTime now) {
        // 查询巴士路线
        List<BusLineCacheDTO> busLineDTOList = busLineService.findBusLineList(ids, now);
        List<Long> lineIds = busLineDTOList.stream().map(item -> item.getId()).collect(Collectors.toList());

        // 查询巴士路线产品
        List<BusProductCacheDTO> busProductDTOS = busProductService.queryBusProductList(lineIds);

        // 将产品设置进路线
        Assist.forEach(busLineDTOList, line -> line.setBusProductList(busProductDTOS.stream().filter(item -> item.getLineId().equals(line.getId())).collect(Collectors.toList())));

        return busLineDTOList;
    }

    /**
     * 巴士订单 - 首页 - 热门路线
     *
     * @return
     */
    @Override
    public Result hotLineQuery(String cityName) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        boolQuery.must(QueryBuilders.matchQuery("departureCityName", cityName == null ? "香港" : cityName));

        boolQuery.must(QueryBuilders.nestedQuery("busProductList", QueryBuilders.matchQuery("busProductList.recommendationStatus", BusRecommendationStatusCode.RECOMMENDATION.getCode()), ScoreMode.None));
        sourceBuilder.query(boolQuery);
        sourceBuilder.sort(new FieldSortBuilder("busProductList.saleCount").order(SortOrder.DESC));
        SearchRequest searchRequest = new SearchRequest(ElasticSearchConstant.BUS_INDEX).source(sourceBuilder);


        List<BusProductCacheDTO> busProductList = new ArrayList<>();
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, ElasticSearchConfiguration.COMMON_OPTIONS);
            SearchHits hits = response.getHits();
            List<BusLineCacheDTO> busLineCacheDTOS = new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
                String sourceAsString = hit.getSourceAsString();
                BusLineCacheDTO busLineCacheDTO = JSON.parseObject(sourceAsString, BusLineCacheDTO.class);
                busLineCacheDTOS.add(busLineCacheDTO);
            }
            if (busLineCacheDTOS.size() > 0) {
                busLineCacheDTOS.forEach(line -> {
                    busProductList.addAll(line.getBusProductList().stream().filter(item -> item.getRecommendationStatus().equals(BusRecommendationStatusCode.RECOMMENDATION.getCode())).collect(Collectors.toList()));
                });
            }
        } catch (IOException e) {
            log.error("【业务模块APP服務-查询es产品列表】异常", e);
            throw new BusinessException(SystemError.ES_IO_EXCEPTION_19004);
        }

        return Result.success(busProductList);
    }

    /**
     * 巴士首页线路查询
     *
     * @param req
     * @return
     */
    @Override
    public Result lineQuery(BusQueryReq req) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        if (req == null) throw new BusinessException(SystemError.BUS_PRODUCT_9003);

        // 出发城市id
        if (req.getDepartureCityId() != null) {
            boolQuery.must(QueryBuilders.matchPhraseQuery("departureCityId", req.getDepartureCityId()));
        }

        // 到达城市id
        if (req.getArriveCityId() != null) {
            boolQuery.must(QueryBuilders.matchPhraseQuery("arriveCityId", req.getArriveCityId()));
        }

        // 上车站id
        if (req.getBoardingStationId() != null) {
            boolQuery.filter(QueryBuilders.matchPhraseQuery("boardingStationId", req.getBoardingStationId()));
        }

        // 下车站id
        if (req.getAlightingStationId() != null) {
            boolQuery.filter(QueryBuilders.matchPhraseQuery("alightingStationId", req.getAlightingStationId()));
        }

        // 出发日期
        LocalTime start = null;
        LocalDate departureDate = req.getDepartureDate();
        if (req.getDepartureDate() != null) {
            if (departureDate.equals(LocalDate.now())) {
                LocalTime now = LocalTime.now();
                start = now.isBefore(LocalTime.of(20, 00)) ? now : LocalTime.of(00, 00);
            }
        }

        // 出发时间段
        if (req.getSatrtTime() != null) {
            start = req.getSatrtTime();
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("departureTime");
            rangeQuery.gte(start);
            if (req.getEndTime() != null) {
                rangeQuery.lte(req.getEndTime());
            }
            boolQuery.filter(rangeQuery);
        }

        SearchRequest searchRequest = new SearchRequest(ElasticSearchConstant.BUS_INDEX).source(sourceBuilder);
        List<BusProductCacheDTO> busProductList = new ArrayList<>();
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, ElasticSearchConfiguration.COMMON_OPTIONS);
            SearchHits hits = response.getHits();
            List<BusLineCacheDTO> busLineCacheDTOS = new ArrayList<>();
            for (SearchHit hit : hits.getHits()) {
                String sourceAsString = hit.getSourceAsString();
                BusLineCacheDTO busLineCacheDTO = JSON.parseObject(sourceAsString, BusLineCacheDTO.class);
                busLineCacheDTOS.add(busLineCacheDTO);
            }
            if (busLineCacheDTOS != null && busLineCacheDTOS.size() > 0) {
                busLineCacheDTOS.forEach(line -> {
                    busProductList.addAll(line.getBusProductList().stream().filter(item -> item.getRecommendationStatus().equals(BusRecommendationStatusCode.RECOMMENDATION.getCode())).collect(Collectors.toList()));
                });
            }
        } catch (IOException e) {
            log.error("【业务模块APP服務-查询es产品列表】异常", e);
            throw new BusinessException(SystemError.ES_IO_EXCEPTION_19004);
        }
        return Result.success(busProductList);
    }

    /**
     * 巴士首页线路查询-到达城市关联
     *
     * @param req
     * @return
     */
    @Override
    public Result arriveCityQuery(BusQueryReq req) {
        SearchSourceBuilder sourceBuilder = new SearchSourceBuilder();
        BoolQueryBuilder boolQuery = QueryBuilders.boolQuery();
        // 入参为空
        if (req == null) throw new BusinessException(SystemError.BUS_PRODUCT_9003);

        // 出发城市id
        if (req.getDepartureCityId() != null) {
            boolQuery.must(QueryBuilders.matchPhraseQuery("departureCityId", req.getDepartureCityId()));
        }

        // 上车站id
        if (req.getBoardingStationId() != null) {
            boolQuery.filter(QueryBuilders.matchPhraseQuery("boardingStationId", req.getBoardingStationId()));
        }

        // 出发日期
        LocalTime start = null;
        LocalDate departureDate = req.getDepartureDate();
        if (req.getDepartureDate() != null) {
            if (departureDate.equals(LocalDate.now())) {
                LocalTime now = LocalTime.now();
                start = now.isBefore(LocalTime.of(20, 00)) ? now : LocalTime.of(00, 00);
            }
        }

        // 出发时间段
        if (req.getSatrtTime() != null) {
            start = req.getSatrtTime();
            RangeQueryBuilder rangeQuery = QueryBuilders.rangeQuery("departureTime");
            rangeQuery.gte(start);
            if (req.getEndTime() != null) {
                rangeQuery.lte(req.getEndTime());
            }
            boolQuery.filter(rangeQuery);
        }

        // 处理响应结果
        SearchRequest searchRequest = new SearchRequest(ElasticSearchConstant.BUS_INDEX).source(sourceBuilder);
        List<ArriveCityDTO> arriveCityDTOS = new ArrayList<>();
        try {
            SearchResponse response = restHighLevelClient.search(searchRequest, ElasticSearchConfiguration.COMMON_OPTIONS);
            SearchHits hits = response.getHits();
            List<BusLineCacheDTO> busLineCacheDTOS = new ArrayList<>();
            // 反序列化命中记录
            for (SearchHit hit : hits.getHits()) {
                String sourceAsString = hit.getSourceAsString();
                BusLineCacheDTO busLineCacheDTO = JSON.parseObject(sourceAsString, BusLineCacheDTO.class);
                busLineCacheDTOS.add(busLineCacheDTO);
            }

            // 去重 - 到达城市名
            Set<String> arriveCitys = new HashSet<>();
            if (busLineCacheDTOS.size() > 0) {
                busLineCacheDTOS.forEach(item -> {
                    arriveCitys.add(item.getArriveCityName());
                });
            }

            // 迭代到达城市，提取站点数据
            Iterator<String> keys = arriveCitys.iterator();
            while (keys.hasNext()) {
                String key = keys.next();

                // 到达城市信息
                List<ArriveCityDTO> arriveCityDTOList = busLineCacheDTOS.stream().map(item -> {
                    ArriveCityDTO arriveCityDTO = new ArriveCityDTO();
                    arriveCityDTO.setArriveCityName(key);
                    if (item.getArriveCityName().equals(key)) {
                        arriveCityDTO.setArriveCityId(item.getArriveCityId());
                        List<ArriveCityDTO.AlightingStationDTO> alightingStationDTOS = new ArrayList<>();

                        // 城市下车站点信息
                        item.getBusProductList().forEach(pro -> {
                            ArriveCityDTO.AlightingStationDTO alightingStationDTO = new ArriveCityDTO.AlightingStationDTO();
                            alightingStationDTO.setAlightingStationId(pro.getAlightingStationId());
                            alightingStationDTO.setAlightingStationName(pro.getAlightingStationName());
                            alightingStationDTOS.add(alightingStationDTO);
                        });
                        // 下车站点添加进城市
                        arriveCityDTO.setAlightingStationDTOS(alightingStationDTOS);
                    }
                    return arriveCityDTO;
                }).collect(Collectors.toList());
                // 到达城市列表
                arriveCityDTOS.addAll(arriveCityDTOList);
            }
        } catch (IOException e) {
            log.error("【业务模块APP服務-查询es产品列表】异常", e);
            throw new BusinessException(SystemError.ES_IO_EXCEPTION_19004);
        }
        return Result.success(arriveCityDTOS);
    }

    /**
     * 巴士缓存更新-按时间
     *
     * @return
     */
    @Override
    public Result busCacheUpdateByTime() {
        up(buildData(null, getCacheUpdate(RedisConstants.BUS_UPDATE_TIME)), RedisConstants.BUS_UPDATE_TIME);
        return Result.success();
    }

    /**
     * 巴士缓存更新-按路线id
     *
     * @param ids
     * @return
     */
    @Override
    public Result busCacheUpdateById(List<Long> ids) {
        up(buildData(ids, null), RedisConstants.BUS_UPDATE_TIME);
        return Result.success();
    }

    /**
     * 设置缓存更新时间
     *
     * @param key
     */
    private void setCacheUpdate(String key) {
        redisTemplate.opsForValue().set(key, JSON.toJSONString(LocalDateTime.now()));
    }

    /**
     * 获取缓存更新时间
     *
     * @param index
     * @return
     */
    private LocalDateTime getCacheUpdate(String index) {
        return JSON.parseObject(redisTemplate.opsForValue().get(index), LocalDateTime.class);
    }

    /**
     * 保存缓存数据
     *
     * @param busLineCacheDTOS
     * @param UpdateTimeKey
     */
    private void up(List<BusLineCacheDTO> busLineCacheDTOS, String UpdateTimeKey) {
        // 保存检索信息到es
        List<BusLineCacheDTO> saveList = busLineCacheDTOS.stream().filter(item -> item.getBusProductList() != null && item.getBusProductList().size() > 0).collect(Collectors.toList());
        List<BusLineCache> busLineCaches = EntityUtil.copyList(saveList, BusLineCache.class);
        busElasticSearchMapper.saveAll(busLineCaches);
        // 标记 上架/更新 时间
        setCacheUpdate(UpdateTimeKey);
        // 保存可售票数到redis
        saveList.forEach(item -> {
            item.getBusProductList().forEach(Product -> {
                Integer inventory = Product.getInventory();
                redisTemplate.opsForValue().set(RedisConstants.BUS_REMAINDER_NUMBER + Product.getId(), inventory.toString());
            });
        });
    }

    /**
     * 执行 K:V 缓存删除
     *
     * @param keys 需要删除的keys
     * @return
     */
    private Map<String, Boolean> deletString(List<String> keys) {
        Map<String, Boolean> map = new HashMap<>();
        keys.forEach(key -> {
            Boolean delete = redisTemplate.delete(key);
            map.put(key, delete);
        });
        return map;
    }

    /**
     * 同步站点主数据
     * @return
     */
    @Override
    public Result<Void> synStopsData() {
    	List<BusStopsCacheDTO> list = busStopsMapper.selectListByKey(BusStopsCacheDTO.class, "status", "0");
    	Assist.forEach(list, dto -> {
    		redisTemplate.opsForSet().add(RedisConstants.BUS_STOPS, toJson(dto));
    	});
    	
		return newResult();
	}
    
    /**
     * 查询缓存站点主数据列表
     */
    @Override
    public Result<List<BusStopsCacheDTO>> queryCacheStopsList(BusStopsDictListQueryReq req) {
    	List<BusStopsCacheDTO> list = queryCacheStopsInfoListHandle(dto -> {
    		if (Assist.isNotBlank(req.getName())) {
    			if (Assist.isBlank(dto.getSiteName()) || !dto.getSiteName().contains(req.getName())) {
    				return false;
    			}
    		}
    		
    		if (Assist.isNotBlank(req.getCityName())) {
    			if (Assist.isBlank(dto.getCityName()) || !dto.getCityName().contains(req.getCityName())) {
    				return false;
    			}
    		}
    		
    		return true;
    	}, BusStopsCacheDTO::getSiteName, req.getPageSize());
    	
    	return newResult(list);
    }
    
    /**
     * 查询缓存城市主数据列表
     */
    @Override
    public Result<List<BusStopsCacheCityDTO>> queryCacheCityList(OperationByLikeReq req) {
    	List<BusStopsCacheDTO> list = queryCacheStopsInfoListHandle(dto -> {
    		if (Assist.isNotBlank(req.getName())) {
    			if (Assist.isBlank(dto.getCityName()) || !dto.getCityName().contains(req.getName())) {
    				return false;
    			}
    		}
    		return true;
    	}, BusStopsCacheDTO::getCityName, req.getPageSize());
    	
    	return newResult(toBeanList(list, BusStopsCacheCityDTO.class));
    }
    
    /**
     * 查询缓存站点相关信息主数据列表逻辑处理
     * @param filterAction 数据过滤方法
     * @param sortedAction 排序方法
     * @param pageSize
     * @return
     */
    private List<BusStopsCacheDTO> queryCacheStopsInfoListHandle(Predicate<BusStopsCacheDTO> filterAction, Function<BusStopsCacheDTO, String> nameAction, int pageSize) {
    	Set<String> jsonList = redisTemplate.opsForSet().members(RedisConstants.BUS_STOPS);
    	List<BusStopsCacheDTO> list = Assist.toJsonBeanList(jsonList, BusStopsCacheDTO.class);
    	//排序
    	list = Assist.sorted(list, nameAction);
    	//临时做法，香港排最前面
    	list = Assist.sorted(list, dto -> {
    		String name = nameAction.apply(dto);
    		if (Assist.isNotBlank(name) && name.contains("香港")) {
    			return 0;
    		}
    		return 1;
    	});
    	
    	//数据过滤
    	List<BusStopsCacheDTO> resultList = Assist.filterToList(list, filterAction);
    	//去重
    	resultList = Assist.distinctToList(resultList, nameAction);

    	return resultList;
    }
}
;